﻿//IQFeedHelper.cs
//Copyright (c) 2013 StockSharp LLC, all rights reserved.
//This code module is part of StockSharp library.
//This code is licensed under the GNU GENERAL PUBLIC LICENSE Version 3.
//See the file License.txt for the license details.
//More info on: http://stocksharp.com

namespace StockSharp.Algo.History.IQFeed
{
	using System;
	using System.Collections.Generic;
	using System.Threading;

	using Ecng.Common;

	static class IQFeedHelper
	{
		public static IQFeedMessage ToMessage(this string message, IQFeedMessageType type)
		{
			switch (type)
			{
				case IQFeedMessageType.Time:
					return new IQFeedTimeMessage(message);
				case IQFeedMessageType.ListedMarket:
					return new IQFeedListedMarketMessage(message);
				case IQFeedMessageType.Security:
					return new IQFeedSecurityMessage(message);
				case IQFeedMessageType.HistoryTrade:
					return new IQFeedHistoryTradeMessage(message);
				case IQFeedMessageType.Quote:
					return new IQFeedQuoteMessage(message);
				case IQFeedMessageType.Trade:
					return new IQFeedTradeMessage(message);
				default:
					return new IQFeedUserMessage(message);
			}
		}

		public static void SyncDo<TIn>(this IQFeedWrapper iqFeed, Action<IQFeedWrapper> action, Action<TIn> func, Action<Exception> processDataError)
			where TIn : IQFeedMessage
		{
			var syncRoot = new object();
			Exception error = null;

			Action<Exception> errorHandler = err =>
			{
				error = err;
				lock (syncRoot)
					Monitor.Pulse(syncRoot);
			};

			Action<IEnumerable<IQFeedMessage>> handler = msgs =>
			{
				foreach (var iqFeedMessage in msgs)
				{
					if (iqFeedMessage.Type == IQFeedMessageType.End)
					{
						lock (syncRoot)
							Monitor.Pulse(syncRoot);
					}
					else if (iqFeedMessage.Type == IQFeedMessageType.Error)
					{
						error = new InvalidOperationException(iqFeedMessage.Value);
						lock (syncRoot)
							Monitor.Pulse(syncRoot);
					}
					else
					{
						iqFeedMessage.DoIf(func);
					}
				}
			};

			try
			{
				iqFeed.ConnectionError += errorHandler;
				iqFeed.ProcessReply += handler;
				iqFeed.ProcessDataError += processDataError;

				iqFeed.Connect();

				action(iqFeed);

				lock (syncRoot)
					Monitor.Wait(syncRoot);

				if (error == null)
					return;

				throw error;
			}
			finally
			{
				iqFeed.Disconnect();

				iqFeed.ProcessDataError -= processDataError;
				iqFeed.ProcessReply -= handler;
				iqFeed.ConnectionError -= errorHandler;
			}
		}

		public static IEnumerable<TOut> SyncGet<TIn, TOut>(this IQFeedWrapper iqFeed, Action<IQFeedWrapper> action, Func<TIn, TOut> func, Action<Exception> processDataError)
			where TIn : IQFeedMessage
		{
			var list = new List<TOut>();

			iqFeed.SyncDo<TIn>(action, m => list.Add(func(m)), processDataError);

			return list;
		}
	}
}
